package com.gbk.clickhouseservicesave.service;

import com.alibaba.fastjson.JSONObject;
import com.gbk.clickhouseservicesave.dto.ChTableData;
import com.les.ads.common.util.JsonUtils;
import com.myweb.ck.ClickHouseTemplate;
import com.myweb.ck.mapping.ChTable;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.event.EventListener;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StopWatch;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import static com.gbk.clickhouseservicesave.task.AsyncTaskConfig.CH_EXE;
import static com.gbk.clickhouseservicesave.task.AsyncTaskConfig.DB_EXE;

/**
 *@ClassName chTableService
 *@Author: gbk
 *@Description TODO:
 *@Version: 1.0
 */

@Service
@Slf4j
public class chTableService {
    @Autowired
    private ClickHouseTemplate clickHouseTemplate;

    @Autowired
    private ApplicationContext applicationContext;

    public static final int DATA_MAX_SIZE = 50000;
    public static final int CLICK_BATCH_SIZE = 30000;
    public static final int PRINT_PRE_MESSAGE = 5000;

    /**
     * 保存数据类型
     */
    private Map<String, Class> chClassMap = new ConcurrentHashMap<>();

    /**
     * 用于保存数据 溢出的次数 ，超过PRINT_PRE_MESSAGE，报一次
     */
    private Map<String, AtomicInteger> overFlowMap = new ConcurrentHashMap<>();

    /**
     * 为每一个ClickHouse 创建一个缓存
     */
    private Map<String, ConcurrentLinkedQueue<Object>> dataCache = new ConcurrentHashMap<>();

    /**
     * 初始化
     */
    @PostConstruct
    public void getAllChTable() {
        Map<String, Object> beansWithAnnotation = applicationContext.getBeansWithAnnotation(ChTable.class);
        if (beansWithAnnotation.isEmpty()) {
            log.info(" !!! ChTable is Empty");
        } else {
            beansWithAnnotation.keySet().forEach(
                    key -> {
                        log.info("clickHouse数据表 --{}--{}", key, beansWithAnnotation.get(key).getClass());
                        dataCache.put(key, new ConcurrentLinkedQueue<>());
                        chClassMap.put(key, beansWithAnnotation.get(key).getClass());
                        overFlowMap.put(key, new AtomicInteger(0));
                    }
            );
        }
    }

    /**
     * 触发监听 填充队列
     */
    @EventListener
    @Async(CH_EXE)
    public void getCHData(ChTableData chTableData) {
        String className = chTableData.getClassName();

        //------------------------------------
        Class clazz = chClassMap.get(className);
        if (clazz == null) {
            log.warn("未知数据类型 --{}", className);
            return;
        }

        ConcurrentLinkedQueue<Object> objectList = dataCache.get(className);
        if (objectList == null) {
            log.warn("未找到数据缓存-{}", className);
            return;
        }

        AtomicInteger atomicInteger = overFlowMap.get(className);
        if (objectList.size() >= DATA_MAX_SIZE) {
            if (atomicInteger.get() % PRINT_PRE_MESSAGE == 0) {
                log.error("");
            }
            atomicInteger.incrementAndGet();
            return;
        }
        atomicInteger.set(0);

        String data = chTableData.getData();
        if (!StringUtils.hasText(data)) {
            log.error("数据项异常 -{}", chTableData);
            return;
        }

        List<Object> objects = JSONObject.parseArray(data, clazz);

        if (CollectionUtils.isEmpty(objects)) {
            log.error("解析数据 -{} -{}异常", clazz.getName(), data);
            return;
        }

        objectList.addAll(objects);
        log.trace("数据项 -{}", objects);
    }

    @Scheduled(initialDelay = 1000, fixedRate = 3000)
    @Async(DB_EXE)
    public void saveData() {
        AtomicBoolean atomicBoolean = new AtomicBoolean(false);
        StopWatch stopWatch = new StopWatch("ch_save");
        for (Map.Entry<String, ConcurrentLinkedQueue<Object>> dataMap : dataCache.entrySet()) {
            stopWatch.start(dataMap.getKey());
            List<Object> list = new ArrayList<>();
            try {
                do {
                    Object object = dataMap.getValue().poll();
                    if (object != null) {
                        list.add(object);
                    } else {
                        break;
                    }
                    if (list.size() >= CLICK_BATCH_SIZE) {
                        break;
                    }
                } while (1 == 1);
                if (!list.isEmpty()) {
                    atomicBoolean.set(true);
                    log.debug("存储数据项 -{}-{}", dataMap.getKey(), list.size());
                    clickHouseTemplate.insertBatch(list);
                }
            } catch (Exception e) {
                log.error("数据存储异常", JsonUtils.toJSONString(list));
                log.error("数据存储异常-{}", e);
                log.error("detail-{}", ExceptionUtils.getRootCause(e));
            } finally {
                stopWatch.stop();
            }
        }
        if (atomicBoolean.get()) {
            log.debug(stopWatch.prettyPrint());
        }

    }


}
